package com.sheepone.distributor;

import com.lmax.disruptor.EventHandler;
import com.sheepone.distributor.event.DistributeEvent;
import com.sheepone.distributor.thread.SafeThreadPool;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Miss.杨
 * @description
 * @since 2023-12-07 22:18
 */
public enum WorkerStationManager implements EventHandler<DistributeEvent> {
    INSTANCE;

    private final ConcurrentHashMap<String, WorkerStation> workerStations = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, SafeThreadPool> threadPools = new ConcurrentHashMap<>();

    public void register(String name, WorkerStation line) {
        workerStations.put(name, line);
    }

    public void registerThreadPool(String station, SafeThreadPool threadPool) {
        threadPools.put(station, threadPool);
    }

    public WorkerStation getWorkerStation(String name) {
        return workerStations.get(name);
    }

    public void stopAll() {
        workerStations.forEach((k, v) -> v.stop());
        threadPools.forEach((k, v) -> v.shotDown());
    }

    @Override
    public void onEvent(DistributeEvent event, long l, boolean b) throws Exception {
        try {
            WorkerStation workerStation = getWorkerStation(event.getTargetStation());
            // 获取对应Topic的listener集合
            // 并封装事件 发布到对应WorkerStation的workerLine中
            workerStation.publish((e, s) -> {
                while(true){
                    if(e.isAvailable() && e.doing()){
                        e.setContext(event.getContext());
                        e.setListeners(workerStation.getEventListeners(event.getTargetTopic()));
                        break;
                    }
                }
            });
        } finally {
            event.finished();
        }
    }
}
